package com.exchange.plugin.kafka;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;


public class TestKafkaConsumer3 {
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		Properties props = new Properties();
    	props.put("bootstrap.servers", "127.0.0.1:9092");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.LongDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("group.id", "getOrderChange");
        props.put("batch.size", 0);
        KafkaConsumer<Long, String> consumer = new KafkaConsumer<Long, String>(props);
        consumer.subscribe(Collections.singletonList("tradeDepth"));//订阅TOPIC
        
        try {
        	
        	Set<TopicPartition> assignment = new HashSet<>();
        	// 在poll()方法内部执行分区分配逻辑，该循环确保分区已被分配。
        	// 当分区消息为0时进入此循环，如果不为0，则说明已经成功分配到了分区。
        	while (assignment.size() == 0) {
        	    consumer.poll(Duration.ofMillis(1000));
        	    // assignment()方法是用来获取消费者所分配到的分区消息的
        	    // assignment的值为：topic-demo-3, topic-demo-0, topic-demo-2, topic-demo-1
        	    assignment = consumer.assignment();
        	}
        	// 指定分区从末尾消费
        	consumer.seekToEnd(assignment);
        	
		    while (true) {
		        // 100 是超时时间（ms），在该时间内 poll 会等待服务器返回数据
		    	@SuppressWarnings("deprecation")
				ConsumerRecords<Long, String> records = consumer.poll(Duration.ofMillis(100)); 
		        // poll 返回一个记录列表。
		        // 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量，以及记录的键值对。(Depth)
		    	//country={"buy":[{"isBuy":true,"number":"0","total":"0","coinTeam":"XBIT-USDT","price":"0","platform":1}],"sell":[{"isBuy":false,"number":"0","total":"0","coinTeam":"XBIT-USDT","price":"0","platform":1}]}
		        for (ConsumerRecord<Long, String> record : records) {
			        System.out.println("customer="+record.key()+", country="+record.value());
		        	System.out.println("交易深度");
			    }
		        /*consumer.commitSync();//同步*/
		    } 
		} finally {
		    // 关闭消费者,网络连接和 socket 也会随之关闭，并立即触发一次再均衡
			consumer.close();
		}
        
        
        
    }
}
